/*
 License
 
 Copyright © 2012 Kevin Kimmich
 
 Permission is hereby granted, free of charge, to any person obtaining a copy
 of this software and associated documentation files (the "Software"), to deal
 in the Software without restriction, including without limitation the rights to 
 use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
 the Software, and to permit persons to whom the Software is furnished to do so, 
 subject to the following conditions:
 
 THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED,
 INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR
 PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE
 FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE,
 ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 */

#include "crcore.h"


static uint32_t chunk_size = (DEFAULT_CHUNK_SIZE);

//allocates and initializes a chunk of memory
chunk_node_t* chunk_malloc( void )
{
	chunk_node_t* chunk = (chunk_node_t*)malloc(sizeof(chunk_node_t));
	memset(chunk,0,sizeof(chunk_node_t));
	chunk->data = malloc(chunk_size);
	return chunk;
}

//frees a chunk of memory
void chunk_free( chunk_node_t* chunk )
{
	free(chunk->data);
	free(chunk);
}

__inline__ void lock_stream(stream* s) {pthread_mutex_lock(&s->stream_mutex);}
__inline__ void unlock_stream(stream* s) {pthread_mutex_unlock(&s->stream_mutex);}

//initialize the stream
//max_size parameter is only used in blocking stream calls
void stream_init( stream* s, uint32_t max_size )
{
	list_init(&s->chunks);
	s->reader_length = START_LENGTH;
	s->writer_length = START_LENGTH;
	s->allocated = 0;
	s->max_size = max_size;
	
	pthread_mutex_init(&s->stream_mutex,NULL);
	pthread_cond_init(&s->read_event.signal,NULL);
	s->read_event.result = 0;
	
	pthread_cond_init(&s->write_event.signal,NULL);
	s->write_event.result = 0;
}

//cleanup the memory used by the stream
void stream_cleanup( stream* s )
{
	lock_stream(s);
	chunk_node_t* chunk;
	for(;;)
	{
		chunk=(chunk_node_t*)s->chunks.head.next;
		if(chunk)
		{
			list_remove(&s->chunks,&chunk->node);
			chunk_free(chunk);
		}
		else
		{
			break;
		}
	}
	
	s->reader_length = START_LENGTH;
	s->writer_length = START_LENGTH;
	s->allocated = 0;
	s->read_event.result = 0;
	s->write_event.result = 0;
	unlock_stream(s);
}

//stream from a file named "filename", return -1 for fail
int stream_from_filename(stream* s, const char* filename)
{
	int fd = open(filename,O_RDONLY);
	if(fd < 0)
		return -1;
	
	struct stat stat;
	if(fstat(fd,&stat))
	{
		close(fd);
		return -1;
	}

	char* mem = mmap(NULL,stat.st_size,PROT_READ,MAP_PRIVATE,fd,0);
	if(mem==NULL)
	{
		close(fd);
		return -1;
	}
	
	stream_from_memory(s,mem,stat.st_size);
	
	munmap(mem,stat.st_size);
	close(fd);
	return stat.st_size;
}

//create a file named "filename" and dump all the stream contents to the file
int stream_to_filename(stream* s, const char* filename)
{
	FILE* f = fopen(filename,"w+");
	if(f)
	{
		io_result result=io_null;
		int bytes_written=0;
		stream_to_file(s,fileno(f),&result,&bytes_written);
		fclose(f);
		return bytes_written;
	}
	return -1;
}

//copy the contents of "data" to the stream up to length bytes
//if data == NULL, fill stream with zeroes.
int stream_from_memory(stream* s, void* data, uint32_t length)
{
	int result=0;
	chunk_node_t* writer;

	lock_stream(s);
	if(s->allocated==0)
	{
		writer = chunk_malloc();
		list_add_tail(&s->chunks,&writer->node);
		s->allocated += chunk_size;
	}
	
	uint32_t data_offset;

	for(data_offset=0;data_offset < length;)
	{
		uint32_t offset = s->writer_length % chunk_size;
		uint32_t chunk_bytes_left = chunk_size - offset;
		uint32_t bytes_to_copy = chunk_bytes_left;
		writer = (chunk_node_t*)s->chunks.head.prev;
		void* stream_data = writer->data + offset;
		
		if((length-data_offset) < bytes_to_copy)
			bytes_to_copy = length-data_offset;
		
		if(data)
			memcpy(stream_data,data + data_offset,bytes_to_copy);
		else
			memset(stream_data,0,bytes_to_copy);
		
		data_offset += bytes_to_copy;
	
		result += bytes_to_copy;
		s->writer_length += bytes_to_copy;
		
		if(bytes_to_copy == chunk_bytes_left)
		{
			list_add_tail(&s->chunks,(list_node_t*)chunk_malloc());
			s->allocated += chunk_size;
		}
	}
	unlock_stream(s);
	return result;
}

//remove the head chunk node and retrun to caller
chunk_node_t* stream_to_chunk(stream* s, int* len, int* last)
{
	chunk_node_t* chunk = NULL;
	
	lock_stream(s);
	if(s->chunks.head.next != NULL)
	{
		uint32_t stream_bytes_left = s->writer_length - s->reader_length;
		
		chunk = (chunk_node_t*)s->chunks.head.next;
		list_remove(&s->chunks,&chunk->node);
		
		if(stream_bytes_left < chunk_size)
		{
			*len=stream_bytes_left;
			*last=1;
			s->allocated -= chunk_size;
			s->reader_length += stream_bytes_left;
		}
		else
		{
			*len=chunk_size;
			*last=0;
			s->allocated -= chunk_size;
			s->reader_length += chunk_size;
		}
	}
	else
	{
		*len=0;
		*last=0;
	}
	unlock_stream(s);
	return chunk;
}

#ifdef ENABLE_LIBXML
//cast an XML document to a stream, this reflects the XML
//centricism of osprey
int doc_to_stream(stream* s, xmlDocPtr doc)
{
	xmlChar* mem = NULL;
	int size = 0;
	xmlDocDumpFormatMemory(doc,&mem,&size,1);
	int rv = stream_from_memory(s,mem,size);
	xmlFree(mem);
	return rv;
}

//cast a stream to an XML document, chunkwise parsing
xmlDocPtr stream_to_doc(stream* s)
{
	xmlDocPtr doc;
	xmlParserCtxtPtr ctx = xmlCreatePushParserCtxt(NULL, NULL, NULL, 0, NULL);
	
	chunk_node_t* chunk;
	int len;
	int last;
	
	for(;;)
	{
		chunk = stream_to_chunk(s,&len,&last);
		if(chunk==NULL)
			break;
		xmlParseChunk(ctx,(const char*)chunk->data,len,last);
		
		chunk_free(chunk);
		
		if(last)
			break;
	}
	
	doc=ctx->myDoc;
	
	if(ctx->lastError.level)
	{
		char errmsg[LOCAL_BUF_SIZE];
		snprintf(errmsg,LOCAL_BUF_SIZE,"%s",ctx->lastError.message);
		while(strstr(errmsg,"\n"))
		{
			char* p = strstr(errmsg,"\n");
			*p='.';
		}
		if(doc)
			xmlFreeDoc(doc);
		doc=NULL;
		xmlFreeParserCtxt(ctx);
		return NULL;
	}
	
	xmlFreeParserCtxt(ctx);
	return doc;
}
#endif
//copy the contents of the stream to a buffer at "data" up to length
//bytes.
int stream_to_memory(stream* s, void* data, uint32_t length)
{
	int result = 0;
	lock_stream(s);
	uint32_t data_offset = 0;
	
	while((s->reader_length != s->writer_length) && (data_offset < length))
	{
		chunk_node_t* reader = (chunk_node_t*)s->chunks.head.next;
		uint32_t stream_bytes_left = (s->writer_length - s->reader_length);
		uint32_t offset = s->reader_length % chunk_size;
		uint32_t chunk_bytes_left = chunk_size - offset;
		void* stream_ptr = reader->data + offset;
		
		uint32_t bytes_to_copy = chunk_bytes_left;
		
		if(stream_bytes_left < chunk_bytes_left)
			bytes_to_copy = stream_bytes_left;
		
		if((length-data_offset) < bytes_to_copy)
			bytes_to_copy = (length-data_offset);
	
		if(data)
			memcpy(data+data_offset,stream_ptr,bytes_to_copy);

		data_offset += bytes_to_copy;
		result += bytes_to_copy;
		s->reader_length += bytes_to_copy;

		if(bytes_to_copy==chunk_bytes_left)
		{
			list_remove(&s->chunks,&reader->node);
			chunk_free(reader);
			s->allocated -= chunk_size;
		}
	}
	unlock_stream(s);
	return result;
}

//returns the contents of a stream as a character buffer terminated
//with a zero
char* stream_to_string(stream* s, int* len)
{
	lock_stream(s);
	uint32_t length = s->writer_length - s->reader_length + 1;
	unlock_stream(s);
	
	char* buf = (char*)malloc(length);
	if(buf==NULL)
		return buf;
	
	buf[length-1]=0;
	stream_to_memory(s,buf,length);
	if(len)
		*len = length;
	return buf;
}

//returns the contents of the stream as a linear buffer
void* stream_to_buffer(stream* s, int* len)
{
	lock_stream(s);
	uint32_t length = s->writer_length - s->reader_length;
	unlock_stream(s);
	
	void* buf = malloc(length);
	if(buf==NULL)
		return buf;
	
	stream_to_memory(s,buf,length);
	if(len)
		*len = length;
	return buf;
}

//dump the contents of source stream to dest stream up to length
int stream_to_stream_length(stream* dst, stream* src, uint32_t length)
{
	int result=0;
	chunk_node_t* writer;
	
	if(length==0)
		return 0;

	lock_stream(dst);
	lock_stream(src);
	
	if(dst->allocated==0)
	{
		writer = chunk_malloc();
		list_add_tail(&dst->chunks,&writer->node);
		dst->allocated += chunk_size;
	}

	while((src->reader_length != src->writer_length) && length)
	{
		writer = (chunk_node_t*)dst->chunks.head.prev;
		uint32_t dst_offset = dst->writer_length % chunk_size;
		uint32_t dst_chunk_bytes_left = chunk_size - dst_offset;
		void* dst_stream_ptr = writer->data + dst_offset;
		
		chunk_node_t* reader = (chunk_node_t*)src->chunks.head.next;
		uint32_t src_offset = src->reader_length % chunk_size;
		uint32_t src_chunk_bytes_left = chunk_size - src_offset;
		void* src_stream_ptr = reader->data + src_offset;
		
		uint32_t src_stream_bytes_left = (src->writer_length - src->reader_length);
		uint32_t bytes_to_copy = dst_chunk_bytes_left;
		
		if(src_stream_bytes_left < bytes_to_copy)
			bytes_to_copy = src_stream_bytes_left;
		
		if(src_chunk_bytes_left < bytes_to_copy)
			bytes_to_copy = src_chunk_bytes_left;
		
		if(length < bytes_to_copy)
			bytes_to_copy = length;
		
		memcpy(dst_stream_ptr,src_stream_ptr,bytes_to_copy);
		result += bytes_to_copy;
		
		if(length != UNBOUNDED)
			length -= bytes_to_copy;
		
		dst->writer_length += bytes_to_copy;
		src->reader_length += bytes_to_copy;
		
		if(bytes_to_copy == dst_chunk_bytes_left)
		{
			list_add_tail(&dst->chunks,(list_node_t*)chunk_malloc());
			dst->allocated += chunk_size;
		}
		
		if(bytes_to_copy == src_chunk_bytes_left)
		{
			list_remove(&src->chunks,&reader->node);
			chunk_free(reader);
			src->allocated -= chunk_size;
		}
	}
	unlock_stream(src);
	unlock_stream(dst);
	return result;
}

struct gzheader {
	unsigned char id1;
	unsigned char id2;
	unsigned char cm;
	unsigned char flg;
	unsigned int mtime;
	unsigned char xfl;
	unsigned char os;
	//unsigned short crc;
} __attribute__((packed));

struct gztrailer {
	unsigned int crc;
	unsigned int len;
};

//dump the src stream contents through gzip to the dst stream
int stream_to_gzip(stream* dst, stream* src)
{
	z_stream strm;
	unsigned char in[DEFAULT_CHUNK_SIZE];
	unsigned char out[4*DEFAULT_CHUNK_SIZE];
	int total_written = 0;
	/* allocate deflate state */
	strm.zalloc = Z_NULL;
	strm.zfree = Z_NULL;
	strm.opaque = Z_NULL;
	
	//(strm, level, method, windowBits, memLevel, strategy)
	int ret = deflateInit2(&strm,9,Z_DEFLATED,-15,9,Z_DEFAULT_STRATEGY);
	if (ret != Z_OK)
		return -1;
	
	struct gzheader hdr = {id1:0x1f,id2:0x8b,cm:8,flg:0,mtime:0,xfl:2,os:3};//,crc:0};
	stream_from_memory(dst,&hdr,sizeof(hdr));
	
	struct gztrailer trl = {0,0};
	trl.crc = crc32(0L, Z_NULL, 0);
	int flush = Z_NO_FLUSH;
	do
	{
		strm.next_in = in;
		strm.avail_in = stream_to_memory(src,in,DEFAULT_CHUNK_SIZE);
		trl.len += strm.avail_in;
		trl.crc = crc32(trl.crc,in,strm.avail_in);
		
		strm.next_out = out;
		strm.avail_out = 4*DEFAULT_CHUNK_SIZE;
		
		if(strm.avail_in >= 0)
		{
			if(strm.avail_in < DEFAULT_CHUNK_SIZE)
				flush=Z_FINISH;
			ret = deflate(&strm,flush);
			
			if(ret != Z_STREAM_ERROR)
			{
				total_written += stream_from_memory(dst,out,4*DEFAULT_CHUNK_SIZE - strm.avail_out);
			}
			else
			{
				return -1;
			}
		}
	}
	while(ret != Z_STREAM_END);
	deflateEnd(&strm);
	
	stream_from_memory(dst,&trl,sizeof(trl));
	
	return total_written;
}

//dump entire src stream to dst stream
int stream_to_stream(stream* dst, stream* src)
{
	return stream_to_stream_length(dst,src,UNBOUNDED);
}

//like vprintf but with the stream as an output
int stream_vprintf(stream* s,const char* format,va_list list)
{
	char buf[LOCAL_BUF_SIZE];
	char* b=buf;
	int length = vsnprintf(b,LOCAL_BUF_SIZE,format,list);
	if(length > LOCAL_BUF_SIZE)
	{
		b = malloc(length+1);
		vsnprintf(b,length+1,format,list);
	}
	
	if(length)
	{
		stream_from_memory(s,b,length);
		if(b!=buf)
			free(b);
		return length;
	}
	return -1;
}

//like printf, but with the stream as an output
int stream_printf(stream* s, const char* format, ...)
{
	va_list list;
	va_start(list,format);
	char buf[LOCAL_BUF_SIZE];
	char* b=buf;
	int length = vsnprintf(b,LOCAL_BUF_SIZE,format,list);
	va_end(list);
	if(length > LOCAL_BUF_SIZE)
	{
		va_list list;
		va_start(list,format);
		b = malloc(length+1);
		vsnprintf(b,length+1,format,list);
		va_end(list);
	}
	
	
	if(length)
	{
		stream_from_memory(s,b,length);
		if(b!=buf)
			free(b);
		return length;
	}
	return -1;
}

//returns the length of the stream
//this should always be used. do not use the writer_length as a replacement
//for the length.
uint32_t stream_length(stream* s)
{
	lock_stream(s);
	uint32_t length = s->writer_length - s->reader_length;
	unlock_stream(s);
	return length;
}

//like strstr()
int stream_strstr(stream* s, const char* needle)
{
	int result;
	lock_stream(s);
	int length = s->writer_length - s->reader_length;
	int needle_length = strlen(needle);
	if(needle_length > length)
	{
		unlock_stream(s);
		return -1;
	}
	chunk_node_t* reader = (chunk_node_t*)s->chunks.head.next;
	
	for(result=0;result<length;result++)	
	{
		uint32_t stream_bytes_left = (s->writer_length - (s->reader_length+result));
		uint32_t offset = (s->reader_length + result) % chunk_size;
		uint32_t chunk_bytes_left = chunk_size - offset;

		if(chunk_bytes_left==0)
		{
			reader=(chunk_node_t*)reader->node.next;
			chunk_bytes_left = chunk_size;
		}
		
		void* stream_ptr = reader->data + offset;
		
		if(stream_bytes_left < needle_length)
			goto notfound;
		
		int cmp_len = chunk_bytes_left;
		
		if(needle_length > cmp_len)
		{
			//two part compare
			if(strncmp(stream_ptr,needle,cmp_len))
				continue;

			stream_bytes_left = (s->writer_length - (s->reader_length+result+cmp_len));
			offset = (s->reader_length + result + cmp_len) % chunk_size;
			chunk_bytes_left = chunk_size - offset;
			stream_ptr = ((chunk_node_t*)reader->node.next)->data + offset;
			
			if(strncmp(stream_ptr,needle+cmp_len,needle_length-cmp_len))
				continue;
			
			goto found;
		}
		else
		{
			//one part compare
			if(strncmp(stream_ptr,needle,needle_length))
				continue;
			
			goto found;
		}
	}
notfound:
	unlock_stream(s);
	return -1;
found:
	unlock_stream(s);
	return result;

}

#ifdef ENABLE_LIBXML
//dump the contents of an XML node to a stream
int node_to_stream(stream* s, xmlNodePtr node, int stream_siblings)
{
	xmlNodePtr n;
	int total=0;
	for(n=node;n;n=n->next)
	{
		if(n->type == XML_TEXT_NODE)
		{
			total += stream_printf(s,(char*)XML_GET_CONTENT(node));
			continue;
		}
		if(n->type == XML_CDATA_SECTION_NODE)
		{
			total += stream_printf(s,"<![CDATA[%s]]>",(char*)XML_GET_CONTENT(node));
			continue;
		}			
		if(n->type != XML_ELEMENT_NODE)
			continue;
		
		total += stream_printf(s,"<%s",(char*)n->name);
		
		xmlNodePtr a;
		for(a	= (xmlNodePtr)n->properties;a;a=a->next)
		{
			char* val = get_node_content(a);
			total += stream_printf(s," %s=\"%s\"",(char*)a->name,val);
		}
		
		if(n->children)
		{
			total += stream_printf(s,">");
			total += node_to_stream(s,n->children,1);
			total += stream_printf(s,"</%s>",n->name);
		}
		else
		{
			total += stream_printf(s,"/>");
		}
		
		if(!stream_siblings)
			return total;
	}
	return total;
}

//dumps the left portion of an XML node
int node_to_stream_left(stream* s, xmlNodePtr node, int no_right)
{
	xmlNodePtr n;
	int total=0;
	for(n=node;n;n=n->next)
	{
		if(n->type == XML_TEXT_NODE)
		{
			total += stream_printf(s,(char*)XML_GET_CONTENT(node));
			continue;
		}			
		if(n->type != XML_ELEMENT_NODE)
			continue;
		
		
		total += stream_printf(s,"<%s",(char*)n->name);
		
		xmlNodePtr a;
		for(a	= (xmlNodePtr)n->properties;a;a=a->next)
		{
			char* val = get_node_content(a);
			total += stream_printf(s," %s=\"%s\"",(char*)a->name,val);
		}
		if(no_right)
			total += stream_printf(s,"/>");
		else
			total += stream_printf(s,">");
		
		return total;
	}
	return total;
}

//dumps the right portion of an XML node
int node_to_stream_right(stream* s, xmlNodePtr node)
{
	xmlNodePtr n;
	int total=0;
	for(n=node;n;n=n->next)
	{
		if(n->type == XML_TEXT_NODE)
		{
			total += stream_printf(s,(char*)XML_GET_CONTENT(node));
			continue;
		}			
		if(n->type != XML_ELEMENT_NODE)
			continue;
		
		total += stream_printf(s,"</%s>",(char*)n->name);
		return total;
	}
	return total;
}

//return a node pointer for a stream
xmlNodePtr stream_to_node(stream* s)
{
	xmlDocPtr doc = stream_to_doc(s);
	if(doc==NULL)
		return NULL;
	xmlNodePtr node = doc->children;
	xmlUnlinkNode(node);
	xmlFreeDoc(doc);
	return node;
}

#endif

//BLOCKING OPERATIONS
//this is a lot cleaner than the streamio interface
int __stream_wait_bytes(stream* s, stream_event* e, int count, int gt_or_lte)
{
	int locked = 0;
	for(;;)
	{
		if(!locked)
			lock_stream(s);
		
		if(e->result==io_closed)
			return -1;
		
		uint32_t len = s->writer_length - s->reader_length;
		
		if(gt_or_lte)
		{
			if(len > count)
			{
				return len;
			}
		}
		else
		{
			if(len <= count)
			{
				return len;
			}
		}
		
		//the mutex is unlocked while waiting
		pthread_cond_wait(&e->signal,&s->stream_mutex);//wait forever
		
		locked=1;
		
		if(e->result==io_closed)
		{	
			return -1;
		}
	}
	return 0;
}

int stream_to_blocking_stream( stream* dst, stream* src )
{
	int result=0;
	chunk_node_t* writer;
	
	lock_stream(dst);
	lock_stream(src);
	
	if(dst->allocated==0)
	{
		writer = chunk_malloc();
		list_add_tail(&dst->chunks,&writer->node);
		dst->allocated += chunk_size;
	}
	
	while((src->reader_length != src->writer_length))
	{
		writer = (chunk_node_t*)dst->chunks.head.prev;
		uint32_t dst_offset = dst->writer_length % chunk_size;
		uint32_t dst_chunk_bytes_left = chunk_size - dst_offset;
		void* dst_stream_ptr = writer->data + dst_offset;
		
		chunk_node_t* reader = (chunk_node_t*)src->chunks.head.next;
		uint32_t src_offset = src->reader_length % chunk_size;
		uint32_t src_chunk_bytes_left = chunk_size - src_offset;
		void* src_stream_ptr = reader->data + src_offset;
		
		uint32_t src_stream_bytes_left = (src->writer_length - src->reader_length);
		uint32_t bytes_to_copy = dst_chunk_bytes_left;
		
		if(src_stream_bytes_left < bytes_to_copy)
			bytes_to_copy = src_stream_bytes_left;
		
		if(src_chunk_bytes_left < bytes_to_copy)
			bytes_to_copy = src_chunk_bytes_left;
		
		memcpy(dst_stream_ptr,src_stream_ptr,bytes_to_copy);
		result += bytes_to_copy;
		
		dst->writer_length += bytes_to_copy;
		src->reader_length += bytes_to_copy;
		
		if(bytes_to_copy == dst_chunk_bytes_left)
		{
			list_add_tail(&dst->chunks,(list_node_t*)chunk_malloc());
			dst->allocated += chunk_size;
		}
		
		if(bytes_to_copy == src_chunk_bytes_left)
		{
			list_remove(&src->chunks,&reader->node);
			chunk_free(reader);
			src->allocated -= chunk_size;
		}
	}
	
	pthread_cond_signal(&dst->read_event.signal);

	unlock_stream(src);
	unlock_stream(dst);
	return result;
}

int stream_blocking_write(stream* s, void* data, uint32_t length)
{
	assert(length);
	
	__stream_wait_bytes(s,&s->write_event,s->max_size,0);
	
	//always locked on return from wait bytes
	
	if(s->write_event.result == io_closed)
	{
		s->write_event.result=io_ok;
		unlock_stream(s);
		return -1;
	}
	
	int result=0;
	chunk_node_t* writer;
	
	if(s->allocated==0)
	{
		writer = chunk_malloc();
		list_add_tail(&s->chunks,&writer->node);
		s->allocated += chunk_size;
	}
	
	uint32_t data_offset;
	
	for(data_offset=0;data_offset < length;)
	{
		uint32_t offset = s->writer_length % chunk_size;
		uint32_t chunk_bytes_left = chunk_size - offset;
		uint32_t bytes_to_copy = chunk_bytes_left;
		writer = (chunk_node_t*)s->chunks.head.prev;
		void* stream_data = writer->data + offset;
		
		if((length-data_offset) < bytes_to_copy)
			bytes_to_copy = length-data_offset;
		
		if(data)
			memcpy(stream_data,data + data_offset,bytes_to_copy);
		else
			memset(stream_data,0,bytes_to_copy);
		
		data_offset += bytes_to_copy;
		
		result += bytes_to_copy;
		s->writer_length += bytes_to_copy;
		
		if(bytes_to_copy == chunk_bytes_left)
		{
			list_add_tail(&s->chunks,(list_node_t*)chunk_malloc());
			s->allocated += chunk_size;
		}
	}
	pthread_cond_signal(&s->read_event.signal);
	unlock_stream(s);
	
	return result;
}

int stream_blocking_read(stream* s, void* data, uint32_t length)
{
	assert(length);
	__stream_wait_bytes(s,&s->read_event,0,1);
	
	if(s->read_event.result==io_closed)
	{
		s->read_event.result=io_ok;
		unlock_stream(s);
		return -1;
	}
	
	int result = 0;
	uint32_t data_offset = 0;
	
	while((s->reader_length != s->writer_length) && (data_offset < length))
	{
		chunk_node_t* reader = (chunk_node_t*)s->chunks.head.next;
		uint32_t stream_bytes_left = (s->writer_length - s->reader_length);
		uint32_t offset = s->reader_length % chunk_size;
		uint32_t chunk_bytes_left = chunk_size - offset;
		void* stream_ptr = reader->data + offset;
		
		uint32_t bytes_to_copy = chunk_bytes_left;
		
		if(stream_bytes_left < chunk_bytes_left)
			bytes_to_copy = stream_bytes_left;
		
		if((length-data_offset) < bytes_to_copy)
			bytes_to_copy = (length-data_offset);
		
		if(data)
			memcpy(data+data_offset,stream_ptr,bytes_to_copy);
		
		data_offset += bytes_to_copy;
		result += bytes_to_copy;
		s->reader_length += bytes_to_copy;
		
		if(bytes_to_copy==chunk_bytes_left)
		{
			list_remove(&s->chunks,&reader->node);
			chunk_free(reader);
			s->allocated -= chunk_size;
		}
	}
	
	pthread_cond_signal(&s->write_event.signal);
	unlock_stream(s);
	
	return result;
}


int __stream_wait_bytes_timeout(stream* s, stream_event* e, int count, int gt_or_lte,uint32_t msec)
{
	int locked=0;
	for(;;)
	{
		if(!locked)
			lock_stream(s);
		
		if(e->result==io_closed)
		{
			return -1;
		}
		
		uint32_t len = s->writer_length - s->reader_length;
		
		if(gt_or_lte)
		{
			if(len > count)
			{
				return len;
			}
		}
		else
		{
			if(len <= count)
			{
				return len;
			}
		}
		
		struct timespec now;
		set_timespec(&now,msec);
		int rv = pthread_cond_timedwait(&e->signal,&s->stream_mutex,&now);
		locked=1;
		
		if(rv)
			return -1;
		
		if(e->result==io_closed)
			return 0;
	}
	
	return 0;
}

int stream_blocking_write_timeout(stream* s, void* data, uint32_t length,uint32_t msec)
{
	if(!length)
		return 0;
	
	if(__stream_wait_bytes_timeout(s,&s->write_event,s->max_size,0,msec)<0)
		return -1;

	if(s->write_event.result == io_closed)
		return -1;
	
	int result=0;
	chunk_node_t* writer;

	if(s->allocated==0)
	{
		writer = chunk_malloc();
		list_add_tail(&s->chunks,&writer->node);
		s->allocated += chunk_size;
	}
	
	uint32_t data_offset;

	for(data_offset=0;data_offset < length;)
	{
		uint32_t offset = s->writer_length % chunk_size;
		uint32_t chunk_bytes_left = chunk_size - offset;
		uint32_t bytes_to_copy = chunk_bytes_left;
		writer = (chunk_node_t*)s->chunks.head.prev;
		void* stream_data = writer->data + offset;
		
		if((length-data_offset) < bytes_to_copy)
			bytes_to_copy = length-data_offset;
		
		if(data)
			memcpy(stream_data,data + data_offset,bytes_to_copy);
		else
			memset(stream_data,0,bytes_to_copy);
		
		data_offset += bytes_to_copy;
	
		result += bytes_to_copy;
		s->writer_length += bytes_to_copy;
		
		if(bytes_to_copy == chunk_bytes_left)
		{
			list_add_tail(&s->chunks,(list_node_t*)chunk_malloc());
			s->allocated += chunk_size;
		}
	}

	pthread_cond_signal(&s->read_event.signal);
	
	unlock_stream(s);
	
	return result;
}


int stream_blocking_read_timeout(stream* s, void* data, uint32_t length,uint32_t msec)
{
	if(!length)
		return 0;
	
	if(__stream_wait_bytes_timeout(s,&s->read_event,0,1,msec)<0)
		return -1;
	
	if(s->read_event.result==io_closed)
		return -1;
	
	int result = 0;
	uint32_t data_offset = 0;
	
	while((s->reader_length != s->writer_length) && (data_offset < length))
	{
		chunk_node_t* reader = (chunk_node_t*)s->chunks.head.next;
		uint32_t stream_bytes_left = (s->writer_length - s->reader_length);
		uint32_t offset = s->reader_length % chunk_size;
		uint32_t chunk_bytes_left = chunk_size - offset;
		void* stream_ptr = reader->data + offset;
		
		uint32_t bytes_to_copy = chunk_bytes_left;
		
		if(stream_bytes_left < chunk_bytes_left)
			bytes_to_copy = stream_bytes_left;
		
		if((length-data_offset) < bytes_to_copy)
			bytes_to_copy = (length-data_offset);
	
		if(data)
			memcpy(data+data_offset,stream_ptr,bytes_to_copy);

		data_offset += bytes_to_copy;
		result += bytes_to_copy;
		s->reader_length += bytes_to_copy;

		if(bytes_to_copy==chunk_bytes_left)
		{
			list_remove(&s->chunks,&reader->node);
			chunk_free(reader);
			s->allocated -= chunk_size;
		}
	}

	pthread_cond_signal(&s->write_event.signal);
	
	unlock_stream(s);
	
	return result;
}

void stream_blocking_cleanup( stream* s )
{
	lock_stream(s);
	chunk_node_t* chunk;
	for(;;)
	{
		chunk=(chunk_node_t*)s->chunks.head.next;
		if(chunk)
		{
			list_remove(&s->chunks,&chunk->node);
			chunk_free(chunk);
		}
		else
		{
			break;
		}
	}
	
	s->reader_length = START_LENGTH;
	s->writer_length = START_LENGTH;
	s->allocated = 0;
	s->read_event.result = io_closed;
	s->write_event.result = io_closed;

	
	pthread_cond_signal(&s->read_event.signal);
	pthread_cond_signal(&s->write_event.signal);
	
	unlock_stream(s);
}

void stream_flush(stream* s)
{
	chunk_node_t* chunk = NULL;
	
	lock_stream(s);
	while(s->chunks.head.next != NULL)
	{
		chunk = (chunk_node_t*)s->chunks.head.next;
		list_remove(&s->chunks,&chunk->node);
		chunk_free(chunk);
	}
	s->reader_length = START_LENGTH;
	s->writer_length = START_LENGTH;
	s->allocated = 0;
	unlock_stream(s);
}

int stream_strcmp(stream* src, stream* dst)
{
	int result=0;
	
	lock_stream(dst);
	lock_stream(src);
	uint32_t slength = src->writer_length - src->reader_length;
	uint32_t dlength = dst->writer_length - dst->reader_length;
	if(slength != dlength)
	{
		unlock_stream(src);
		unlock_stream(dst);
		return 1;
	}
	
	uint32_t sreader_length=src->reader_length;
	uint32_t dreader_length=dst->reader_length;
	
	chunk_node_t* reader = (chunk_node_t*)src->chunks.head.next;
	chunk_node_t* writer = (chunk_node_t*)dst->chunks.head.next;
	
	while(sreader_length != src->writer_length)
	{
		uint32_t dst_offset = dreader_length % chunk_size;
		uint32_t dst_chunk_bytes_left = chunk_size - dst_offset;
		void* dst_stream_ptr = writer->data + dst_offset;
		
		uint32_t src_offset = sreader_length % chunk_size;
		uint32_t src_chunk_bytes_left = chunk_size - src_offset;
		void* src_stream_ptr = reader->data + src_offset;
		
		uint32_t src_stream_bytes_left = (src->writer_length - sreader_length);
		uint32_t bytes_to_copy = dst_chunk_bytes_left;
		
		if(src_stream_bytes_left < bytes_to_copy)
			bytes_to_copy = src_stream_bytes_left;
		
		if(src_chunk_bytes_left < bytes_to_copy)
			bytes_to_copy = src_chunk_bytes_left;
		
		if(dst_chunk_bytes_left < bytes_to_copy)
			bytes_to_copy = dst_chunk_bytes_left;
		
		result = memcmp(dst_stream_ptr,src_stream_ptr,bytes_to_copy);
		
		if(result)
			break;
		
		dreader_length += bytes_to_copy;
		sreader_length += bytes_to_copy;
		
		if(bytes_to_copy == dst_chunk_bytes_left)
		{
			writer = (chunk_node_t*)writer->node.next;
		}
		
		if(bytes_to_copy == src_chunk_bytes_left)
		{
			reader = (chunk_node_t*)reader->node.next;
		}
	}
	unlock_stream(src);
	unlock_stream(dst);
	return result;
}

char* stream_to_string_keep(stream* src, int* len)
{
	int result=0;
	lock_stream(src);
	
	uint32_t length = src->writer_length - src->reader_length + 1;
	if(!length)
	{
		if(len) *len=0;
		unlock_stream(src);
		return NULL;
	}
	char* buf = malloc(length);
	buf[length-1]=0;
	
	uint32_t reader_length = src->reader_length;
	chunk_node_t* reader = (chunk_node_t*)src->chunks.head.next;
	uint32_t dst_offset=0;
	
	while(reader_length != src->writer_length)
	{
		void* dst_stream_ptr = buf + dst_offset;
		
		uint32_t src_offset = reader_length % chunk_size;
		uint32_t src_chunk_bytes_left = chunk_size - src_offset;
		void* src_stream_ptr = reader->data + src_offset;
		
		uint32_t src_stream_bytes_left = (src->writer_length - reader_length);
		uint32_t bytes_to_copy = src_chunk_bytes_left;
		if(src_stream_bytes_left < bytes_to_copy)
			bytes_to_copy=src_stream_bytes_left;
		
		memcpy(dst_stream_ptr,src_stream_ptr,bytes_to_copy);
		result += bytes_to_copy;
		dst_offset += bytes_to_copy;
		reader_length += bytes_to_copy;
		
		reader = (chunk_node_t*)reader->node.next;
	}
	unlock_stream(src);
	if(len)
		*len=result;
	return buf;
}

static void __stream_cleanup(stream* s)
{
	chunk_node_t* chunk;
	for(;;)
	{
		chunk=(chunk_node_t*)s->chunks.head.next;
		if(chunk)
		{
			list_remove(&s->chunks,&chunk->node);
			chunk_free(chunk);
		}
		else
		{
			break;
		}
	}
	
	s->reader_length = START_LENGTH;
	s->writer_length = START_LENGTH;
	s->allocated = 0;
	s->read_event.result = 0;
	s->write_event.result = 0;
}

int stream_duplicate(stream* src, stream* dst)
{
	int result=0;
	chunk_node_t* writer;
	
	lock_stream(dst);
	lock_stream(src);
	
	__stream_cleanup(dst);
	writer = chunk_malloc();
	list_add_tail(&dst->chunks,&writer->node);
	dst->allocated += chunk_size;
	
	uint32_t reader_length = src->reader_length;
	chunk_node_t* reader = (chunk_node_t*)src->chunks.head.next;
	
	while(reader_length != src->writer_length)
	{
		uint32_t dst_offset = dst->writer_length % chunk_size;
		uint32_t dst_chunk_bytes_left = chunk_size - dst_offset;
		void* dst_stream_ptr = writer->data + dst_offset;
		
		uint32_t src_offset = reader_length % chunk_size;
		uint32_t src_chunk_bytes_left = chunk_size - src_offset;
		void* src_stream_ptr = reader->data + src_offset;
		
		uint32_t src_stream_bytes_left = (src->writer_length - reader_length);
		uint32_t bytes_to_copy = dst_chunk_bytes_left;
		
		if(src_stream_bytes_left < bytes_to_copy)
			bytes_to_copy = src_stream_bytes_left;
		
		if(src_chunk_bytes_left < bytes_to_copy)
			bytes_to_copy = src_chunk_bytes_left;
		
		if(dst_chunk_bytes_left < bytes_to_copy)
			bytes_to_copy = dst_chunk_bytes_left; 
		
		memcpy(dst_stream_ptr,src_stream_ptr,bytes_to_copy);
		result += bytes_to_copy;
		
		dst->writer_length += bytes_to_copy;
		reader_length += bytes_to_copy;
		
		if(bytes_to_copy == dst_chunk_bytes_left)
		{
			writer = chunk_malloc();
			list_add_tail(&dst->chunks,&writer->node);
			dst->allocated += chunk_size;
		}
		
		if(bytes_to_copy == src_chunk_bytes_left)
		{
			reader = (chunk_node_t*)reader->node.next;
		}
	}
	unlock_stream(src);
	unlock_stream(dst);
	return result;
}

